#include "zrpc/net/connector.h"

namespace zrpc {
const int Connector::kMaxRetryDelayMs;

Connector::Connector(EventLoop *loop_, const char *ip, int16_t port, bool retry)
    : loop_(loop_),
      ip_(ip),
      port_(port),
      retry_(retry),
      state_(kDisconnected),
      connect_(false),
      retry_delay_ms_(kInitRetryDelayMs) {}

Connector::~Connector() { assert(!channel_); }

void Connector::Start(bool state) {
  connect_ = true;
  loop_->RunInLoop(std::bind(&Connector::StartInLoop, this, state));
}

void Connector::StartInLoop(bool state) {
  loop_->AssertInLoopThread();
  assert(state_ == kDisconnected);
  if (connect_) {
    Connecting(state);
  } else {
    LOG_WARN << "do not connect_";
  }
}

void Connector::Stop() {
  connect_ = false;
  loop_->QueueInLoop(std::bind(&Connector::StopInLoop, this));
}

void Connector::StopInLoop() {
  loop_->AssertInLoopThread();
  if (state_ == kConnecting) {
    SetState(kDisconnected);
    RemoveAndResetChannel();
  }
}

void Connector::ResetChannel() { channel_.reset(); }

int32_t Connector::RemoveAndResetChannel() {
  channel_->DisableAll();
  channel_->Remove();
  int32_t sockfd = channel_->Getfd();
  loop_->QueueInLoop(std::bind(&Connector::ResetChannel, this));
  return sockfd;
}

void Connector::Connecting(bool state, int32_t sockfd) {
  if (state) {
    SetState(kConnected);
    if (connect_) {
      socket::SetSocketNonBlock(sockfd);
      new_connection_callback_(sockfd);
    } else {
      socket::Close(sockfd);
    }
  } else {
    assert(!channel_);
    channel_.reset(new Channel(loop_, sockfd));
    channel_->SetWriteCallback(std::bind(&Connector::HandleWrite, this));
    channel_->SetErrorCallback(std::bind(&Connector::HandleError, this));
    channel_->EnableWriting();
  }
}

void Connector::Retry(int32_t sockfd) {
  socket::Close(sockfd);
  SetState(kDisconnected);
  if (!retry_) {
    return;
  }

  if (connect_) {
    LOG_WARN << "Connector::Retry - Retry Connecting to " << ip_ << " " << port_
             << " in " << retry_delay_ms_ << " milliseconds. ";
    loop_->RunAfter(
        retry_delay_ms_ / 1000.0, false,
        std::bind(&Connector::StartInLoop, shared_from_this(), false));
#ifdef _WIN64
    retry_delay_ms_ = (retry_delay_ms_ * 2) < (kMaxRetryDelayMs)
                          ? (retry_delay_ms_ * 2)
                          : (kMaxRetryDelayMs);
#else
    retry_delay_ms_ = std::min(retry_delay_ms_ * 2, kMaxRetryDelayMs);
#endif
  } else {
    LOG_DEBUG << "do not connect_";
  }
}

void Connector::Restart() {
  loop_->AssertInLoopThread();
  SetState(kDisconnected);
  retry_delay_ms_ = kInitRetryDelayMs;
  connect_ = true;
  StartInLoop(false);
}

void Connector::HandleWrite() {
  if (state_ == kConnecting) {
    int sockfd = RemoveAndResetChannel();
    int err = socket::GetSocketError(sockfd);
    if (err) {
      Retry(sockfd);
    } else if (socket::IsSelfConnect(sockfd)) {
      Retry(sockfd);
    } else {
      SetState(kConnected);
      if (connect_) {
        new_connection_callback_(sockfd);
      } else {
        socket::Close(sockfd);
      }
    }
  } else {
    assert(state_ == kDisconnected);
  }
}

void Connector::HandleError() {
  if (state_ == kConnecting) {
    int sockfd = RemoveAndResetChannel();
    int err = socket::GetSocketError(sockfd);
    LOG_TRACE << "SO_ERROR = " << err << " " << strerror(err);
    Retry(sockfd);
  }
}

void Connector::Connecting(bool state) {
  int32_t sockfd;
  if (state) {
    sockfd = socket::CreateSocket();
  } else {
    sockfd = socket::CreateNonblockingOrDie();
  }

  int32_t ret = socket::Connect(sockfd, ip_.c_str(), port_);
  int32_t saved_errno = (ret == 0) ? 0 : errno;

  switch (saved_errno) {
    case 0:
    case EINPROGRESS:
    case EINTR:
    case EISCONN:
      SetState(kConnecting);
      Connecting(state, sockfd);
      break;
    case EAGAIN:
    case EADDRINUSE:
    case EADDRNOTAVAIL:
    case ECONNREFUSED:
    case ENETUNREACH:
      Retry(sockfd);
      break;
    case EACCES:
    case EPERM:
    case EAFNOSUPPORT:
    case EALREADY:
    case EBADF:
    case EFAULT:
    case ENOTSOCK:
      Retry(sockfd);
      break;
    default:
      Retry(sockfd);
      break;
  }
}

}  // namespace zrpc